package com.silky.officeline.thread;


import com.silky.common.bean.DataSourceConfig;
import com.silky.common.bean.MultiLevelJoinConfig;
import com.silky.common.bean.NodeConfig;
import com.silky.common.constant.IConstant;
import com.silky.common.service.NodeService;
import com.silky.common.utils.PageList;
import com.silky.common.utils.ThreadSafeList;
import lombok.extern.slf4j.Slf4j;

import java.sql.Connection;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

/**
 *  历史数据多级单线程
 */
@Slf4j
public class MultiLevelSingleThread  extends AbstractThread{
    private final DataSourceConfig dataSourceConfig;
    private final MultiLevelJoinConfig multiLevelJoinConfig;
    private final PageList initPageList;
    private final CountDownLatch countDownLatch;
    private final NodeService nodeService;
    private final int currentLevel;
    private Connection connection=null;
    public MultiLevelSingleThread(DataSourceConfig dataSourceConfig, MultiLevelJoinConfig multiLevelJoinConfig, PageList initPageList, CountDownLatch countDownLatch, int currentLevel) {
        this.dataSourceConfig = dataSourceConfig;
        this.multiLevelJoinConfig = multiLevelJoinConfig;
        this.initPageList = initPageList;
        this.countDownLatch = countDownLatch;
        this.nodeService = new NodeService(dataSourceConfig);
        this.currentLevel = currentLevel;
    }

    @Override
    public void run()
    {
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"开始执行");
        try {
            connection = nodeService.getConnectionByDataSourceName(multiLevelJoinConfig.getNodes().get(currentLevel).getDataSourceName());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("获取数据库连接失败",e);
            System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"退出");
            return;
        }
        while (true) {
            //从页数列表中获取一个页数
            Long offset = initPageList.remove();
            if(offset==null) {
                break;
            }
            List<Map<String, Object>> currentNodeDatas = null;
            NodeConfig currentNodeConfig = multiLevelJoinConfig.getNodes().get(currentLevel);
            try {
                currentNodeDatas = nodeService.getTableData(currentNodeConfig, offset, connection);
                //debug
                traceLostData(currentNodeDatas);
            } catch (Exception e) {
                log.error("获取当前级别数据失败",e);
                e.printStackTrace();
                break;
            }
            if(currentLevel+1<=multiLevelJoinConfig.getNodes().size()-1){
                NodeConfig nextNode = multiLevelJoinConfig.getNodes().get(currentLevel+1);
                if(nextNode.getType().equals(IConstant.NODE_TYPE_JOIN)){
                    try {
                        processNextLevelJoin(currentNodeDatas,nextNode,currentLevel+1,currentNodeConfig.getAlias());
                    } catch (Exception e) {
                        e.printStackTrace();
                        log.error("获取当前级别数据失败",e);
                        break;
                    }
                }else if(nextNode.getType().equals(IConstant.NODE_TYPE_UPDATE)){
                    try {
                        processNextLevelUpdate(currentNodeDatas,nextNode,currentLevel+1,currentNodeConfig.getAlias());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        log.error("更新下级数据失败",e);
                        break;
                    }
                }else if (nextNode.getType().equals(IConstant.NODE_TYPE_SINGLE)){
                    log.warn("没有意义，忽略！");
                }

            }else{
                log.warn("已经没有下级节点了，当前级别为{},节点数组长度为{}",currentLevel,multiLevelJoinConfig.getNodes().size());
                break;
            }
        }
        //当前的CountDownlatch减1
        countDownLatch.countDown();
        if(connection!=null){
            nodeService.closeConnection(connection);
        }
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"退出");
    }

    private void traceLostData(List<Map<String, Object>> currentNodeDatas) {
        for(Map<String,Object> data:currentNodeDatas){
            String schemeId = data.get("schemeId").toString();
            if(schemeId.equals("2479355402210705408")||
               schemeId.equals("7244032309306888192")||
               schemeId.equals("7307746228067995648")){
               System.out.println("丢失数据:"+schemeId);
            }
        }
    }

    /**
     * 执行下级节点为Update节点的情况
     * @param currentNodeDatas
     * @param nextNode
     * @param nextLevel
     * @param alias
     */
    private void processNextLevelUpdate(List<Map<String, Object>> currentNodeDatas, NodeConfig nextNode, int nextLevel, String alias) throws InterruptedException {
        ThreadSafeList threadSafeList = new ThreadSafeList();
        threadSafeList.init(currentNodeDatas);
        CountDownLatch currentCountDownLatch = new CountDownLatch(threadSafeList.size());
        for(int i=0;i<multiLevelJoinConfig.getThreadNum();i++) {
            MultiLevelUpdateThread multiLevelUpdateThread = new MultiLevelUpdateThread(dataSourceConfig, multiLevelJoinConfig, threadSafeList, nextNode, nextLevel, currentCountDownLatch);
            multiLevelUpdateThread.start();
        }
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"等待processNextLevelUpdate下级线程完成");
        currentCountDownLatch.await();
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"processNextLevelUpdate下级线程完成");
    }

    /**
     *  处理下级节点为Join节点的情况
     * @param currentNodeDatas
     * @param nextNode
     * @param level
     * @param alias currentNodeDatas
     */
    private void processNextLevelJoin(List<Map<String, Object>> currentNodeDatas, NodeConfig nextNode, int level, String alias) throws Exception {
        //计算下一节点页数列表
        PageList nextPageList = new PageList();
        long totalCount = nodeService.getCount(nextNode);
        List<Long> offsets = nodeService.splitOffset(totalCount, nextNode.getBatchSize());
        nextPageList.init(offsets);
        //构造下级CountDownLatch
        CountDownLatch nextCountDownLatch = new CountDownLatch(multiLevelJoinConfig.getThreadNum());
        Thread[] threads = new Thread[multiLevelJoinConfig.getThreadNum()];
        for (int i = 0; i < threads.length; i++) {
            threads[i] = new Thread(new MultiLevelJoinThread(dataSourceConfig,multiLevelJoinConfig,currentNodeDatas,nextPageList,nextCountDownLatch,level,alias));
            threads[i].start();
        }
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"等待processNextLevelJoin下级线程完成,nextPageList.size()="+nextPageList.size());
        //等待子线程处理完毕
        nextCountDownLatch.await();
        System.out.println("MultiLevelSingleThread当前线程："+Thread.currentThread().getName()+"processNextLevelJoin下级线程处理完成");
    }
}
